package com.demo.controller;

import cn.hutool.core.util.StrUtil;
import com.demo.entity.User;
import com.demo.enums.Role;
import com.demo.service.IUserService;
import lombok.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.client.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.ArrayList;

/**
 *
 * https://blog.csdn.net/CSDN2497242041/article/details/148592376
 * @author gy
 * @date 2022/6/15
 */
@RestController
@RequestMapping("/api")
public class WebFluxTestController {

    WebClient webClientBuilder =  WebClient.builder()
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .build();

    WebClient webClient = WebClient.create("http://localhost:8080");

//    WebClient client1 = WebClient.builder()
//            .filter(filterA).filter(filterB).build();
//
//    WebClient client2 = client1.mutate()
//            .filter(filterC).filter(filterD).build();

    WebClient client = WebClient.create("https://example.org");

//    Mono<ResponseEntity<Person>> result = client.get()
//            .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
//            .retrieve()
//            .toEntity(Person.class);

//    Flux<Quote> result = client.get()
//            .uri("/quotes").accept(MediaType.TEXT_EVENT_STREAM)
//            .retrieve()
//            .bodyToFlux(Quote.class);

    @RequestMapping("/hello")
    public String hello(HttpServletRequest request, HttpServletResponse response){
        return "hello" + request.getRequestURL();
    }


    @RequestMapping("/mono")
    public Mono<String> mono(HttpServletRequest request, HttpServletResponse response){
        return Mono.just("hello" + request.getRequestURL());
    }

    @RequestMapping("/flux")
    public Flux<String> flux(HttpServletRequest request, HttpServletResponse response){
        ArrayList<String> list = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            list.add("hello" + i);
        }
        Flux<String> flux = Flux.fromIterable(list);
        return flux;
    }


    @RequestMapping("/body")
    public void test5(){
        WebClient webClient = WebClient.create("http://localhost:8080/api/restful");

        MultiValueMap<String, String> formData = new LinkedMultiValueMap<>();
        formData.add("name", "张和");
        formData.add("color", "blue");

        Mono<Dog> dogMono = webClient.post()
                .contentType(MediaType.APPLICATION_FORM_URLENCODED)
                .bodyValue(formData)
                .retrieve()
                .bodyToMono(Dog.class);
        System.out.println(dogMono.block());
    }

    @RequestMapping("/body1")
    public void test6(){
        WebClient webClient = WebClient.create("http://localhost:8080/api/json");

        Mono<Dog> mono = Mono.just(new Dog("和", "33"));

        Mono<Dog> dogMono = webClient.post()
                .contentType(MediaType.APPLICATION_JSON)
                .body(mono, Dog.class)
                .retrieve()
                .bodyToMono(Dog.class);
        System.out.println(dogMono.block());
    }

    public void test7(){

        Mono<Dog> mono = Mono.just(new Dog("和", "33"));

        WebClient webClient = WebClient.builder().filter(new ExchangeFilterFunction() {
            @Override
            public Mono<ClientResponse> filter(ClientRequest clientRequest, ExchangeFunction exchangeFunction) {
                ClientRequest build = ClientRequest.from(clientRequest).body(mono, Dog.class).build();
                return exchangeFunction.exchange(build);
            }
        }).build();

        Mono<Dog> dogMono = webClient.post()
                .uri("http://localhost:8080/api/json")
                .contentType(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToMono(Dog.class);
        System.out.println(dogMono.block());
    }

//    @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
//    public Flux<Event> getEvents() {
//        return eventService.getEventStream();
//    }

    public void test8(){
        webClient.get()
                .uri("/users/1")
                .exchange()
                .flatMap(response -> {
                    if (response.statusCode().is2xxSuccessful()) {
                        return response.bodyToMono(User.class);
                    } else {
                        return Mono.error(new RuntimeException("Request failed with status code: " + response.statusCode()));
                    }
                })
                .subscribe(user -> System.out.println("User: " + user));
    }

    public void test9(){
        webClient.get()
                .uri("/stream")
                .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)// 发送的是 JSON 数据
                .accept(MediaType.TEXT_EVENT_STREAM)    //  希望返回流式数据
                .exchange()
                .flatMapMany(clientResponse -> clientResponse.bodyToFlux(String.class))
                .subscribe(data -> System.out.println("Received: " + data));
    }

    /**
     * 4.5 exchange() 与 .retrieve() 的对比
     * 特性	.retrieve()	.exchange()
     * 响应处理	自动处理 2xx 成功响应	手动处理所有响应
     * 错误处理	默认抛出异常	需要手动判断状态码或错误
     * 获取响应头	不支持	支持
     * 流式响应	支持但不够灵活	完全支持，可自由控制
     * 使用难度	简单	复杂但更强大
     * 4.6 适用场景
     * 场景	推荐方法
     * 普通 GET 请求获取 JSON 数据	.retrieve()
     * 需要判断不同状态码	.exchange()
     * 处理 SSE 或流式响应	.exchange()
     * 需要访问响应头信息	.exchange()
     * 自定义错误逻辑（如重试、降级）	.exchange()
     * ————————————————
     * 版权声明：本文为CSDN博主「AI何哥」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
     * 原文链接：https://blog.csdn.net/CSDN2497242041/article/details/148592376
     */
    public void test10(){
        webClient.get()
                .uri("/info")
                .exchange()
                .flatMap(response -> {
                    String contentType = response.headers().contentType().map(MediaType::toString).orElse("unknown");
                    return response.bodyToMono(String.class)
                            .map(body -> "Content-Type: " + contentType + ", Body: " + body);
                })
                .subscribe(System.out::println);
    }

    public void test11(){
        //Flux<CommonResult<AiChatMessageSendRespVO>>
//        return streamResponse.map(chunk -> {
//            String newContent = chunk.getResult() != null ? chunk.getResult().getOutput().getContent() : null;
//            newContent = StrUtil.nullToDefault(newContent, ""); // 避免 null 的 情况
//            contentBuffer.append(newContent);
//            // 响应结果
//            return success(new AiChatMessageSendRespVO()
//                    .setSend(BeanUtils.toBean(userMessage, AiChatMessageSendRespVO.Message.class))
//                    .setReceive(BeanUtils.toBean(assistantMessage, AiChatMessageSendRespVO.Message.class)
//                            .setContent(newContent))
//            );
//        }).doOnComplete(() -> {
//            // 忽略租户，因为 Flux 异步无法透传租户
//            TenantUtils.executeIgnore(() ->
//                    chatMessageMapper.updateById(new AiChatMessageDO()
//                            .setId(assistantMessage.getId())
//                            .setSegmentIds(convertList(segmentList, AiKnowledgeSegmentDO::getId))
//                            .setContent(contentBuffer.toString())));
//        }).doOnError(throwable -> {
//            log.error("[sendChatMessageStream][userId({}) sendReqVO({}) 发生异常]", finalUserId, sendReqVO, throwable);
//            // 忽略租户，因为 Flux 异步无法透传租户
//            TenantUtils.executeIgnore(() ->
//                    chatMessageMapper.updateById(new AiChatMessageDO().setId(assistantMessage.getId()).setContent(throwable.getMessage())));
//        }).onErrorResume(error -> Flux.just(error(AiErrorCodeConstants.CHAT_STREAM_ERROR)));
    }

    /**
     * 同步创建固定数据流
     * 数据流未执行直到订阅
     */
    public void test12(){
        Flux.just("A",  "B", "C"); // 创建包含3个字符串的流
        Mono.just("Hello");        // 创建包含单个字符串的 Mono

//        调用 just() 仅是声明数据流，不会触发元素发送。必须通过 subscribe() 订阅后才会实际发出数据：
        Mono.just("Data").subscribe(System.out::println); // 输出 "Data"

        Mono.empty();
        Flux.empty();

        Mono<String> mono = Mono.justOrEmpty(null); // 不会发射任何数据，直接 onComplete

        Flux<String> flux = Flux.just("Apple", "Banana", "Cherry");

        flux.subscribe(System.out::println);

        String[] fruits = {"Apple", "Banana", "Cherry"};
        Flux<String> flux1 = Flux.just(fruits);

    }

    @GetMapping("/user")
    public Mono<User> getUser() {
        User user = new User(1,"Alice",20);
        return Mono.just(user); // 包装成响应式类型返回
    }

    @GetMapping("/numbers")
    public Flux<Integer> getNumbers() {
        return Flux.just(1, 2, 3, 4, 5); // 返回一个数字序列
    }

    public void test13(){
        Mono<String> mono = Mono.just("Hello World");

        // 在此之前，"Hello World" 并没有被打印出来，因为还没有订阅

        mono.subscribe(System.out::println); // 订阅后，输出 "Hello World"

        Flux<String> flux = Flux.just("A", "B", "C");

        // 同样地，这里不会立即打印任何内容

        flux.subscribe(System.out::println); // 订阅后，依次输出 "A", "B", "C"
    }

    public void test14(){
        /**
         * 1）flatMap特点
         * 异步非阻塞
         *
         * 并发执行多个内部流
         *
         * 常用于处理异步数据转换、数据库查询、HTTP 请求等
         *
         * 可控制并发数（通过 flatMap(..., concurrency)）
         *
         * （2）对比 map()方法
         * map() 是同步转换。
         *
         * flatMap() 是异步映射 + 合并
         */
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .flatMap(i -> Flux.range(i * 10, 3)); // 每个数字变成一个范围流

        // 输出：10, 11, 12, 20, 21, 22, 30, 31, 32
        flux.subscribe(System.out::println);
    }

    public void test15(){
        Flux<Integer> flux = Flux.generate(
                () -> 0, // 初始状态
                (state, sink) -> {
                    sink.next(state); // 发送当前状态
                    if (state == 10) {
                        sink.complete(); // 完成流
                    }
                    return state + 1; // 返回新的状态
                }
        );

        flux.subscribe(System.out::println);
    }

    public void test16(){
        Flux<String> flux = Flux.create(sink -> {
            sink.next("Hello");
            sink.next("World");
            sink.complete();
        });

        flux.subscribe(System.out::println);
    }

    public void test17(){
        Mono<String> mono = Mono.fromSupplier(() -> "Hello from supplier");

        mono.subscribe(System.out::println);
    }

    @Data
    class Dog{
        private String name;
        private String color;
        public Dog(String name, String color) {
            this.name = name;
            this.color = color;
        }
    }

}
